package cloud.zhixinglu.www.spring.boot.paho.mqtt.mqtt;

import cn.hutool.core.util.IdUtil;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;

/**
 * MQTT生产者客户端
 * @author  xct
 * @date  2021/7/30 15:32
 */
@Slf4j
@Configuration
public class MqttProviderConfig {

	@Value("${mqtt.url}")
	private String hostUrl;

	@Value("${mqtt.username}")
	private String username;

	@Value("${mqtt.password}")
	private String password;

	@Value("${mqtt.qos}")
	private Integer qos;

	@Value("${mqtt.retained}")
	private boolean retained;

	@Value("${mqtt.connectionTimeout}")
	private Integer connectionTimeout;

	@Value("${mqtt.keepAliveInterval}")
	private Integer keepAliveInterval;

	@Value("${mqtt.providerPrefix}")
	private String providerPrefix;

	/**
	 * 客户端对象
	 */
	private MqttClient client;

	/**
	 * 在bean初始化后连接到服务器
	 * @author xct
	 * @date 2021/7/30 16:48
	 */
	@PostConstruct
	public void init(){
		connect();
	}

	/**
	 * 客户端连接服务端
	 * @author xct
	 * @date 2021/7/30 16:01
	 */
	public void connect(){
		String clientId = providerPrefix + "_" + IdUtil.getSnowflakeNextIdStr();
		log.info("MqttProviderConfig.connect info, provider client id :{}", clientId);
		try {
			// 创建MQTT客户端对象
			client = new MqttClient(hostUrl, clientId, new MemoryPersistence());
			// 连接设置
			MqttConnectOptions options = createMqttConnectOptions();
			// 设置回调
			client.setCallback(new MqttProviderCallBack());
			client.connect(options);
		} catch (MqttException e) {
			e.printStackTrace();
			log.error("MqttProviderConfig.connect error, mqtt connection failed, error: {}", e.getMessage());
		}
	}

    /**
     * 创建 MQTT 链接配置
	 * @return MqttConnectOptions
	 */
	private MqttConnectOptions createMqttConnectOptions() {
		MqttConnectOptions options = new MqttConnectOptions();
		// 是否自动重连
		options.setAutomaticReconnect(true);
		// 重连间隔时间的最大值，默认为 128000 毫秒
		options.setMaxReconnectDelay(128000);
		// 是否清空session，设置为false表示服务器会保留客户端的连接记录（订阅主题，qos），客户端重连之后能获取到服务器在客户端断开连接期间推送的消息
		// 设置为true表示每次连接到服务端都是以新的身份
		options.setCleanSession(true);
		// 设置连接用户名
		options.setUserName(username);
		// 设置连接密码
		options.setPassword(password.toCharArray());
		// 设置超时时间，单位为秒
		options.setConnectionTimeout(connectionTimeout);
		// 设置心跳时间 单位为秒，表示服务器每隔1.5*20秒的时间向客户端发送心跳判断客户端是否在线
		options.setKeepAliveInterval(keepAliveInterval);
		// 遗嘱消息，当客户端异常断开时发送的消息
		return options;
	}

	/**
	 * 服务端发送消息
	 * @param topic String 主题
	 * @param message String 消息
	 */
	public void publish(String topic, String message){
		publish(qos, retained, topic, message);
	}

	/**
	 *  服务端发送消息
	 * @param retained boolean 是否保留消息
	 * @param topic String 主题
	 * @param message String 消息
	 */
	public void publish(boolean retained, String topic, String message){
		publish(qos, retained, topic, message);
	}

	/**
	 * 服务端发送消息
	 * @param qos int
	 * @param retained boolean 是否保留消息
	 * @param topic String 主题
	 * @param message String 消息
	 */
	public void publish(int qos, boolean retained, String topic, String message){
		MqttMessage mqttMessage = new MqttMessage();
		mqttMessage.setQos(qos);
		mqttMessage.setRetained(retained);
		mqttMessage.setPayload(message.getBytes());
    	// 主题目的地，用于发布/订阅消息
		MqttTopic mqttTopic = client.getTopic(topic);

		//提供一种机制来跟踪消息的传递进度。
		//用于在以非阻塞方式（在后台运行）执行发布时跟踪消息的传递进度
		MqttDeliveryToken token;
		try {
			//将指定消息发布到主题，但不等待消息传递完成。返回的token可用于跟踪消息的传递状态。
			//一旦此方法干净地返回，消息就已被客户端接受发布。当连接可用时，将在后台完成消息传递。
			token = mqttTopic.publish(mqttMessage);
			token.waitForCompletion();
		} catch (MqttException e) {
			e.printStackTrace();
			log.error("MqttProviderConfig.publish error, mqtt publishing failed, error: {}", e.getMessage());
		}
	}
}
